package stream

import "io"

// Proxy calls recvSend until it receives an error. The error is returned
// to the caller unless it is io.EOF.
func Proxy(revSend func() error) (err error) {
	for err == nil {
		err = revSend()
	}

	if err == io.EOF {
		err = nil
	}
	return err
}

// CloseSender captures the CloseSend method from gRPC streams.
type CloseSender interface {
	CloseSend() error
}

// ProxyFast works like Proxy but runs multiple callbacks simultaneously.
// It returns immediately if proxying one of the callbacks fails. If the
// response stream is done, ProxyBidi returns immediately without waiting
// for the client stream to finish proxying.
func ProxyFast(requestFunc func() error, requestStream CloseSender, responseFunc func() error) error {
	requestErr := make(chan error, 1)
	go func() {
		requestErr <- requestFunc()
	}()

	responseErr := make(chan error, 1)
	go func() {
		responseErr <- responseFunc()
	}()

	for {
		select {
		case err := <-requestErr:
			if err != nil {
				return err
			}
			err = requestStream.CloseSend()
			if err != nil {
				return err
			}
		case err := <-responseErr:
			return err
		}
	}
}

func ProxyLoop(requestFunc func() error, requestStream CloseSender, responseFunc func() error) error {
	requestErr := make(chan error, 1)
	go func() {
		requestErr <- Proxy(requestFunc)
	}()

	responseErr := make(chan error, 1)
	go func() {
		responseErr <- Proxy(responseFunc)
	}()

	for {
		select {
		case err := <-requestErr:
			if err != nil {
				return err
			}
			err = requestStream.CloseSend()
			if err != nil {
				return err
			}
		case err := <-responseErr:
			return err
		}
	}
}
